package com.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.BaseAppV2;
import com.atguigu.gmall.realtime.bean.OrderWide;
import com.atguigu.gmall.realtime.bean.PaymentWide;
import com.atguigu.gmall.realtime.bean.ProductStats;
import com.atguigu.gmall.realtime.common.Constant;
import com.atguigu.gmall.realtime.function.DimAsyncFunction;
import com.atguigu.gmall.realtime.util.AtguiguUtil;
import com.atguigu.gmall.realtime.util.DimUtil;
import com.atguigu.gmall.realtime.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import redis.clients.jedis.Jedis;

import java.sql.Connection;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Collections;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import static com.atguigu.gmall.realtime.common.Constant.*;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/9/1 9:43
 */
public class DwsProduct extends BaseAppV2 {
    public static void main(String[] args) {
        new DwsProduct().init(
            4002, 1, "DwsProduct", "DwsProduct",
            TOPIC_DWD_PAGE, TOPIC_DWD_DISPLAY,
            TOPIC_DWD_FAVOR_INFO, TOPIC_DWD_CART_INFO,
            TOPIC_DWM_ORDER_WIDE, TOPIC_DWM_PAYMENT_WIDE,
            TOPIC_DWD_ORDER_REFUND_INFO, TOPIC_DWD_COMMENT_INFO);
        
    }
    
    @Override
    protected void run(StreamExecutionEnvironment env, Map<String, DataStreamSource<String>> streamMap) {
        // 1. 解析, union成一个流
        DataStream<ProductStats> productStatsStream = parseAndUnionOne(streamMap);
        // 2. 开窗聚合
        SingleOutputStreamOperator<ProductStats> aggregatedStream = aggregate(productStatsStream);
        
        // 3. join维度
        SingleOutputStreamOperator<ProductStats> psStream = joinDim(aggregatedStream);
        
        // 4. 写入到clickhouse中
        writeToClickhouse(psStream);
        
        // 5. 数据向kafka写一份, 给 商品行为关键词主题宽表使用
        writeToKafka(psStream);
    
    }
    
    private void writeToKafka(SingleOutputStreamOperator<ProductStats> psStream) {
        psStream
            .map(JSON::toJSONString)
            .addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWS_PRODUCT_STATS));
    }
    
    private void writeToClickhouse(SingleOutputStreamOperator<ProductStats> psStream) {
        psStream.addSink(FlinkSinkUtil.getClickhouseSink(CLICKHOUSE_DB, CLICKHOUSE_TABLE_PRODUCT_STATS, ProductStats.class));
    }
    
    private SingleOutputStreamOperator<ProductStats> joinDim(SingleOutputStreamOperator<ProductStats> aggregatedStream) {
      return  AsyncDataStream.unorderedWait(
            aggregatedStream,
            new DimAsyncFunction<ProductStats>() {
                @Override
                protected void addDim(Connection conn, 
                                      Jedis redis,
                                      ProductStats ps,
                                      ResultFuture<ProductStats> resultFuture) {
                    // 3. 补充sku
                    JSONObject skuInfo = DimUtil.getDim(conn, redis, "dim_sku_info", ps.getSku_id());
                    ps.setSku_name(skuInfo.getString("SKU_NAME"));
                    ps.setSku_price(skuInfo.getBigDecimal("PRICE"));
    
                    ps.setSpu_id(skuInfo.getLong("SPU_ID"));
                    ps.setCategory3_id(skuInfo.getLong("CATEGORY3_ID"));
                    ps.setTm_id(skuInfo.getLong("TM_ID"));
                    // 4. 补充spu
                    JSONObject spuInfo = DimUtil.getDim(conn, redis, "dim_spu_info", ps.getSpu_id());
                    ps.setSpu_name(spuInfo.getString("SPU_NAME"));
    
                    // 5. 补充 tm
                    JSONObject tmInfo = DimUtil.getDim(conn, redis, "dim_base_trademark", ps.getTm_id());
                    ps.setTm_name(tmInfo.getString("TM_NAME"));
                    // 5. 补充 c3
                    JSONObject c3Info = DimUtil.getDim(conn, redis, "dim_base_category3", ps.getCategory3_id());
                    ps.setCategory3_name(c3Info.getString("NAME"));
    
                    resultFuture.complete(Collections.singletonList(ps));
                    
                }
            },
            60,
            TimeUnit.SECONDS
        );
    }
    
    private SingleOutputStreamOperator<ProductStats> aggregate(DataStream<ProductStats> productStatsStream) {
        return productStatsStream
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<ProductStats>forBoundedOutOfOrderness(Duration.ofSeconds(10))
                    .withTimestampAssigner((ps, ts) -> ps.getTs())
            )
            .keyBy(ProductStats::getSku_id)
            .window(TumblingEventTimeWindows.of(Time.seconds(5)))
            .reduce(
                new ReduceFunction<ProductStats>() {
                    @Override
                    public ProductStats reduce(ProductStats ps1, ProductStats ps2) throws Exception {
                        ps1.setClick_ct(ps1.getClick_ct() + ps2.getClick_ct());
                        ps1.setDisplay_ct(ps1.getDisplay_ct() + ps2.getDisplay_ct());
                        
                        ps1.setFavor_ct(ps1.getFavor_ct() + ps2.getFavor_ct());
                        ps1.setCart_ct(ps1.getCart_ct() + ps2.getCart_ct());
                        
                        ps1.setOrder_amount(ps1.getOrder_amount().add(ps2.getOrder_amount()));
                        ps1.setOrder_sku_num(ps1.getOrder_sku_num() + ps2.getOrder_sku_num());
                        ps1.getOrderIdSet().addAll(ps2.getOrderIdSet());  // 把ps2中所有的订单id放入ps1的HashSet中
                        
                        ps1.setPayment_amount(ps1.getPayment_amount().add(ps2.getPayment_amount()));
                        ps1.getPaidOrderIdSet().addAll(ps2.getPaidOrderIdSet());  // 把ps2中所有的订单id放入ps1的HashSet中
                        
                        ps1.setRefund_amount(ps1.getRefund_amount().add(ps2.getRefund_amount()));
                        ps1.getRefundOrderIdSet().addAll(ps2.getRefundOrderIdSet());
                        
                        ps1.setComment_ct(ps1.getComment_ct() + ps2.getComment_ct());
                        ps1.setGood_comment_ct(ps1.getGood_comment_ct() + ps2.getGood_comment_ct());
                        
                        return ps1;
                    }
                },
                new ProcessWindowFunction<ProductStats, ProductStats, Long, TimeWindow>() {
                    
                    private SimpleDateFormat sdf;
                    
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                    }
                    
                    @Override
                    public void process(Long key,
                                        Context ctx,
                                        Iterable<ProductStats> elements,
                                        Collector<ProductStats> out) throws Exception {
                        ProductStats ps = elements.iterator().next();
                        
                        ps.setStt(sdf.format(ctx.window().getStart()));
                        ps.setEdt(sdf.format(ctx.window().getEnd()));
                        
                        ps.setOrder_ct((long) ps.getOrderIdSet().size());
                        ps.setPaid_order_ct((long) ps.getPaidOrderIdSet().size());
                        ps.setRefund_order_ct((long) ps.getRefundOrderIdSet().size());
                        
                        out.collect(ps);
                    }
                }
            );
        
    }
    
    private DataStream<ProductStats> parseAndUnionOne(Map<String, DataStreamSource<String>> streamMap) {
        // 1. 点击量
        SingleOutputStreamOperator<ProductStats> clickStatsStream = streamMap
            .get(TOPIC_DWD_PAGE)
            .flatMap(new FlatMapFunction<String, ProductStats>() {
                @Override
                public void flatMap(String value,
                                    Collector<ProductStats> out) throws Exception {
                    JSONObject obj = JSON.parseObject(value);
                    JSONObject page = obj.getJSONObject("page");
                    
                    String pageId = page.getString("page_id");
                    String itemType = page.getString("item_type");
                    if ("good_detail".equals(pageId) && "sku_id".equals(itemType)) {
                        Long item = page.getLong("item");
                        ProductStats ps = new ProductStats();
                        ps.setSku_id(item);
                        ps.setClick_ct(1L);
                        ps.setTs(obj.getLong("ts"));
                        out.collect(ps);
                    }
                    
                }
            });
        
        // 2. 曝光流
        SingleOutputStreamOperator<ProductStats> displayStatsStream = streamMap
            .get(TOPIC_DWD_DISPLAY)
            .map(new MapFunction<String, ProductStats>() {
                @Override
                public ProductStats map(String value) throws Exception {
                    JSONObject display = JSON.parseObject(value);
                    Long skuId = display.getLong("item");
                    Long ts = display.getLong("ts");
                    ProductStats ps = new ProductStats();
                    ps.setSku_id(skuId);
                    ps.setDisplay_ct(1L);
                    ps.setTs(ts);
                    return ps;
                }
            });
        
        // 3. 收藏
        SingleOutputStreamOperator<ProductStats> favorStatsSteam = streamMap
            .get(TOPIC_DWD_FAVOR_INFO)
            .map(new MapFunction<String, ProductStats>() {
                @Override
                public ProductStats map(String value) throws Exception {
                    JSONObject favor = JSON.parseObject(value);
                    Long skuId = favor.getLong("sku_id");
                    long ts = AtguiguUtil.toTs(favor.getString("create_time"));
                    
                    ProductStats ps = new ProductStats();
                    ps.setSku_id(skuId);
                    ps.setFavor_ct(1L);
                    ps.setTs(ts);
                    return ps;
                }
            });
        
        // 4. 加购物车
        SingleOutputStreamOperator<ProductStats> cartStatsStream = streamMap
            .get(TOPIC_DWD_CART_INFO)
            .map(new MapFunction<String, ProductStats>() {
                @Override
                public ProductStats map(String value) throws Exception {
                    JSONObject cart = JSON.parseObject(value);
                    Long skuId = cart.getLong("sku_id");
                    long ts = AtguiguUtil.toTs(cart.getString("create_time"));
                    
                    ProductStats ps = new ProductStats();
                    ps.setSku_id(skuId);
                    ps.setCart_ct(1L);
                    ps.setTs(ts);
                    return ps;
                }
            });
        
        // 5. 计算订单
        SingleOutputStreamOperator<ProductStats> orderStatsStream = streamMap
            .get(TOPIC_DWM_ORDER_WIDE)
            .map(new MapFunction<String, ProductStats>() {
                @Override
                public ProductStats map(String value) throws Exception {
                    OrderWide orderWide = JSON.parseObject(value, OrderWide.class);
                    
                    ProductStats ps = new ProductStats();
                    ps.setSku_id(orderWide.getSku_id());
                    ps.getOrderIdSet().add(orderWide.getOrder_id());
                    
                    ps.setOrder_amount(orderWide.getSplit_total_amount());
                    ps.setOrder_sku_num(orderWide.getSku_num());
                    
                    ps.setTs(AtguiguUtil.toTs(orderWide.getCreate_time()));
                    return ps;
                }
            });
        
        // 6. 支付订单
        SingleOutputStreamOperator<ProductStats> payStatsStream = streamMap
            .get(TOPIC_DWM_PAYMENT_WIDE)
            .map(new MapFunction<String, ProductStats>() {
                @Override
                public ProductStats map(String value) throws Exception {
                    PaymentWide paymentWide = JSON.parseObject(value, PaymentWide.class);
                    
                    ProductStats ps = new ProductStats();
                    ps.setSku_id(paymentWide.getSku_id());
                    ps.getPaidOrderIdSet().add(paymentWide.getOrder_id());
                    
                    ps.setPayment_amount(paymentWide.getSplit_total_amount());
                    
                    ps.setTs(AtguiguUtil.toTs(paymentWide.getPayment_create_time()));
                    return ps;
                }
            });
        
        // 7. 退款
        SingleOutputStreamOperator<ProductStats> refundStatsStream = streamMap
            .get(TOPIC_DWD_ORDER_REFUND_INFO)
            .map(new MapFunction<String, ProductStats>() {
                @Override
                public ProductStats map(String value) throws Exception {
                    JSONObject refund = JSON.parseObject(value);
                    Long skuId = refund.getLong("sku_id");
                    long ts = AtguiguUtil.toTs(refund.getString("create_time"));
                    
                    ProductStats ps = new ProductStats();
                    ps.setSku_id(skuId);
                    ps.setTs(ts);
                    
                    ps.setRefund_amount(refund.getBigDecimal("refund_amount"));
                    ps.getRefundOrderIdSet().add(refund.getLong("order_id"));
                    return ps;
                }
            });
        
        // 8. 评价
        SingleOutputStreamOperator<ProductStats> commentStatsStream = streamMap
            .get(TOPIC_DWD_COMMENT_INFO)
            .map(new MapFunction<String, ProductStats>() {
                @Override
                public ProductStats map(String value) throws Exception {
                    JSONObject comment = JSON.parseObject(value);
                    Long skuId = comment.getLong("sku_id");
                    long ts = AtguiguUtil.toTs(comment.getString("create_time"));
                    
                    ProductStats ps = new ProductStats();
                    ps.setSku_id(skuId);
                    ps.setTs(ts);
                    
                    ps.setComment_ct(1L);
                    
                    String appraise = comment.getString("appraise");
                    
                    if (FIVE_STAR_COMMENT.equals(appraise) || FOUR_STAR_COMMENT.equals(appraise)) {
                        ps.setGood_comment_ct(1L);
                        
                    }
                    return ps;
                }
            });
        
        return clickStatsStream.union(
            displayStatsStream,
            favorStatsSteam,
            cartStatsStream,
            orderStatsStream,
            payStatsStream,
            refundStatsStream,
            commentStatsStream
        );
        
    }
}
